package com.lock;

import javafx.fxml.LoadException;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class ZKLock implements Lock, Watcher {

    private static final String ROOT_LOCK = "/locks";
    private static final int sessionTimeout = 30000;

    private ZooKeeper zk = null;
    /**
     * 竞争的资源
     */
    private String lockName;
    private String WAIT_LOCK;
    private String CURRENT_LOCK;


    private CountDownLatch latch;

    private List<Exception> exceptions = new LinkedList<>();

    public ZKLock(String config, String lockName) {
        this.lockName = lockName;
        try {
            zk = new ZooKeeper(config, sessionTimeout, this);
            Stat stat = zk.exists(ROOT_LOCK, false);
            if (stat == null) {
                //不存在，创建
                zk.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        if (!exceptions.isEmpty()) {
            throw new LockException(exceptions.get(0));
        }
        if (tryLock()) {
            return;
        } else {
            //等待锁
            try {
                waitForLock(WAIT_LOCK, sessionTimeout);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        this.lock();
    }

    @Override
    public boolean tryLock() {
        String splitStr = "_lock_";
        if (lockName.contains(splitStr)) {
            throw new LockException("锁名有误");
        }
        //创建临时有序节点
        try {
            CURRENT_LOCK = zk.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("CURRENT_LOCK:" + CURRENT_LOCK + "已经创建");
            //获取所有的子节点
            List<String> subNodes = zk.getChildren(ROOT_LOCK, false);
            //取出所有lockName的锁
            List<String> lockObjects = new ArrayList<>();
            for (String node : subNodes) {
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName)) {
                    lockObjects.add(node);
                }
            }
            Collections.sort(lockObjects);
            System.out.println(Thread.currentThread().getName() + "的锁是 " + CURRENT_LOCK);
            //若当前节点为最小节点，则获取成功
            if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
                System.out.println(Thread.currentThread().getName() + "获取锁成功 " + CURRENT_LOCK);
                return true;
            }

            //若不是最小节点，则找自己前一个节点
            String preNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
            WAIT_LOCK = lockObjects.get(Collections.binarySearch(lockObjects, preNode) - 1);
            System.out.println(Thread.currentThread().getName() + " 需要等待锁： WAIT_LOCK:" + WAIT_LOCK);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (tryLock()) {
            return true;
        }
        try {
            return waitForLock(WAIT_LOCK, time);
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public void unlock() {
        System.out.println(Thread.currentThread().getName() + " 释放锁 " + CURRENT_LOCK);
        try {
            zk.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void process(WatchedEvent event) {
        if (this.latch != null) {
            this.latch.countDown();
        }
    }

    private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(ROOT_LOCK + "/" + prev, true);
        if (stat != null) {
            System.out.println(Thread.currentThread().getName() + "等待锁 " + ROOT_LOCK + "/" + prev);
            this.latch = new CountDownLatch(1);
            this.latch.await(waitTime, TimeUnit.MILLISECONDS);
            this.latch = null;
            System.out.println(Thread.currentThread().getName() + "等到了锁");
        }
        return true;
    }
}
